# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: site-packages\requests_ntlm\requests_ntlm.py
import binascii, sys, warnings
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.exceptions import UnsupportedAlgorithm
from ntlm_auth import ntlm
from requests.auth import AuthBase
from requests.packages.urllib3.response import HTTPResponse


class HttpNtlmAuth(AuthBase):
    """
    HTTP NTLM Authentication Handler for Requests.
    
    Supports pass-the-hash.
    """

    def __init__(self, username, password, session=None, send_cbt=True):
        r"""Create an authentication handler for NTLM over HTTP.
        
        :param str username: Username in 'domain\username' format
        :param str password: Password
        :param str session: Unused. Kept for backwards-compatibility.
        :param bool send_cbt: Will send the channel bindings over a HTTPS channel (Default: True)
        """
        if ntlm is None:
            raise Exception("NTLM libraries unavailable")
        try:
            self.domain, self.username = username.split("\\", 1)
        except ValueError:
            self.username = username
            self.domain = ""

        if self.domain:
            self.domain = self.domain.upper()
        self.password = password
        self.send_cbt = send_cbt
        self.session_security = None

    def retry_using_http_NTLM_auth(
        self, auth_header_field, auth_header, response, auth_type, args
    ):
        server_certificate_hash = self._get_server_cert(response)
        if auth_header in response.request.headers:
            return response
        content_length = int(
            response.request.headers.get("Content-Length", "0"), base=10
        )
        if hasattr(response.request.body, "seek"):
            if content_length > 0:
                response.request.body.seek(-content_length, 1)
            else:
                response.request.body.seek(0, 0)
            response.content
            response.raw.release_conn()
            request = response.request.copy()
            context = ntlm.Ntlm()
            negotiate_message = context.create_negotiate_message(self.domain).decode(
                "ascii"
            )
            auth = "%s %s" % (auth_type, negotiate_message)
            request.headers[auth_header] = auth
            args_nostream = dict(args, stream=False)
            response2 = response.connection.send(request, **args_nostream)
            response2.content
            response2.raw.release_conn()
            request = response2.request.copy()
            if response2.headers.get("set-cookie"):
                request.headers["Cookie"] = response2.headers.get("set-cookie")
            auth_header_value = response2.headers[auth_header_field]
            auth_strip = auth_type + " "
            ntlm_header_value = next(
                (
                    s
                    for s in (val.lstrip() for val in auth_header_value.split(","))
                    if s.startswith(auth_strip)
                )
            ).strip()
            context.parse_challenge_message(ntlm_header_value[len(auth_strip) :])
            authenticate_message = context.create_authenticate_message(
                self.username,
                self.password,
                self.domain,
                server_certificate_hash=server_certificate_hash,
            )
            authenticate_message = authenticate_message.decode("ascii")
            auth = "%s %s" % (auth_type, authenticate_message)
            request.headers[auth_header] = auth
            response3 = response2.connection.send(request, **args)
            response3.history.append(response)
            response3.history.append(response2)
            self.session_security = context.session_security
            return response3

    def response_hook(self, r, **kwargs):
        """The actual hook handler."""
        if r.status_code == 401:
            www_authenticate = r.headers.get("www-authenticate", "").lower()
            auth_type = _auth_type_from_header(www_authenticate)
            if auth_type is not None:
                return self.retry_using_http_NTLM_auth(
                    "www-authenticate", "Authorization", r, auth_type, kwargs
                )
        else:
            if r.status_code == 407:
                proxy_authenticate = r.headers.get("proxy-authenticate", "").lower()
                auth_type = _auth_type_from_header(proxy_authenticate)
                if auth_type is not None:
                    return self.retry_using_http_NTLM_auth(
                        "proxy-authenticate",
                        "Proxy-authorization",
                        r,
                        auth_type,
                        kwargs,
                    )
                return r

    def _get_server_cert(self, response):
        """
        Get the certificate at the request_url and return it as a hash. Will get the raw socket from the
        original response from the server. This socket is then checked if it is an SSL socket and then used to
        get the hash of the certificate. The certificate hash is then used with NTLMv2 authentication for
        Channel Binding Tokens support. If the raw object is not a urllib3 HTTPReponse (default with requests)
        then no certificate will be returned.
        
        :param response: The original 401 response from the server
        :return: The hash of the DER encoded certificate at the request_url or None if not a HTTPS endpoint
        """
        if self.send_cbt:
            certificate_hash = None
            raw_response = response.raw
            if isinstance(raw_response, HTTPResponse):
                if sys.version_info > (3, 0):
                    socket = raw_response._fp.fp.raw._sock
                else:
                    socket = raw_response._fp.fp._sock
                try:
                    server_certificate = socket.getpeercert(True)
                except AttributeError:
                    pass

                certificate_hash = _get_certificate_hash(server_certificate)
            else:
                warnings.warn(
                    "Requests is running with a non urllib3 backend, cannot retrieve server certificate for CBT",
                    NoCertificateRetrievedWarning,
                )
            return certificate_hash
        else:
            return

    def __call__(self, r):
        r.headers["Connection"] = "Keep-Alive"
        r.register_hook("response", self.response_hook)
        return r


def _auth_type_from_header(header):
    """
    Given a WWW-Authenticate or Proxy-Authenticate header, returns the
    authentication type to use. We prefer NTLM over Negotiate if the server
    suppports it.
    """
    if "ntlm" in header:
        return "NTLM"
    if "negotiate" in header:
        return "Negotiate"


def _get_certificate_hash(certificate_der):
    cert = x509.load_der_x509_certificate(certificate_der, default_backend())
    try:
        hash_algorithm = cert.signature_hash_algorithm
    except UnsupportedAlgorithm as ex:
        warnings.warn(
            "Failed to get signature algorithm from certificate, unable to pass channel bindings: %s"
            % str(ex),
            UnknownSignatureAlgorithmOID,
        )
        return

    if hash_algorithm.name in ("md5", "sha1"):
        digest = hashes.Hash(hashes.SHA256(), default_backend())
    else:
        digest = hashes.Hash(hash_algorithm, default_backend())
    digest.update(certificate_der)
    certificate_hash_bytes = digest.finalize()
    certificate_hash = binascii.hexlify(certificate_hash_bytes).decode().upper()
    return certificate_hash


class NoCertificateRetrievedWarning(Warning):
    pass


class UnknownSignatureAlgorithmOID(Warning):
    pass
